package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo19PageRank {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("pageRank")
      .setMaster("local")

    val sc = new SparkContext(conf)

    val lineRDD: RDD[String] = sc.textFile("data/pageRank.txt")

    var linesRDD: RDD[(String, List[String], Double)] = lineRDD.map(line => {

      val lines: Array[String] = line.split(" ")

      val page: String = lines(0)

      val list: List[String] = lines(1).split(",").toList

      (page,list,1.0)
    })

    //网页编号和出链列表
    val pageListRDD: RDD[(String, List[String])] = linesRDD.map(kv => (kv._1,kv._2))


    var i = 0
    var flag = true
    //网页总数目
    val pageNum: Long = pageListRDD.count()
    //阻尼系数
    val q: Double = 0.85

    while (flag){

      var listValueRDD: RDD[(String, Double)] = linesRDD.flatMap(s => {
        //取出出链列表
        val list: List[String] = s._2
        //计算出出链页面的平均值
        val value: Double = s._3 / list.length
        //取每个页面的出链，并返回出链页面和平均值
        val lv: List[(String, Double)] = list.map(ss => {
          (ss,value)
        })
        lv
      })

      //reduceByKey：计算各个页面的PR和，mapValues：修正PageRank值
      val resultRDD: RDD[(String, Double)] = listValueRDD.reduceByKey(_ + _).mapValues(s => (1-q) / pageNum + q*s)

      //将新求出的PageRank值通过jion操作更新上一次得到的PageRank值
      val joinRDD: RDD[(String, (List[String], Double))] = pageListRDD.join(resultRDD)

      //处理jion得到的数据
      val newLines: RDD[(String, List[String], Double)] = joinRDD.map(s => {
        val page: String = s._1
        //取出链和pr值
        val listAndValue: (List[String], Double) = s._2
        //返回页面，对应的出链，pr值
        (page,listAndValue._1,listAndValue._2)
      })

      //newLines.foreach(println)

      /**
        * 设定一个差值指标（0.0001）。当所有页面和上一次计算的PR差值平均小于该标准时，则收敛。
        *
        */
      //获取上一次的各个网页的PR值
      val lastListPR: RDD[(String, Double)] = linesRDD.map(kv => (kv._1,kv._3))
      //获取当前各个网页的PR值
      val nowListPR: RDD[(String, Double)] = newLines.map(kv =>(kv._1,kv._3))
      //join
      val joinPR: RDD[(String, (Double, Double))] = lastListPR.join(nowListPR)
      //计算PR的差值
      val prChaRDD: RDD[Double] = joinPR.map{
        case ((_: String, (lastPR: Double, nowPR: Double))) =>
          Math.abs(nowPR - lastPR)
      }
      //计算平均值
      val prAvg: Double = prChaRDD.sum() / prChaRDD.count()

      //判断
      if(prAvg < 0.0001){
        flag = false
      }

      //更新PageRank值
      linesRDD = newLines

      i+=1
    }
    linesRDD.foreach(println)
    println("循环次数为："+i)


  }

}
